package com.sdy.mq.base;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

@Slf4j
public class BaseMessageListener implements MessageListenerConcurrently {
    private BaseMessageConsumer messageConsumer;
    
    public BaseMessageListener(BaseMessageConsumer messageConsumer) {
        this.messageConsumer = messageConsumer;
    }
    
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        try {
            for (MessageExt message : msgs) {
                log.debug("Receive message: Topic=[{}],Tags=[{}],MsgId=[{}],Keys=[{}]", message.getTopic(), message.getTags(), message.getMsgId(), message.getKeys());
                if (!messageConsumer.consume(message)) {
                    log.error("Consumer messages failed, reconsume later.");
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }
        } catch (Exception e) {
            log.error("Consumer messages exception, reconsume later.", e);
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}
